#pragma once
#ifndef __linux__
#error unix local stream server only support linux system
#endif

#include <string>
#include <set>
#include <functional>
#include <memory>				//shared_ptr use
#include <queue>
#include <atomic>
#ifndef ASIO_STANDALONE
#define ASIO_STANDALONE
#define ASIO_NO_DEPRECATED
#endif
#include "asio/io_context.hpp" 	//io_context use
#include "asio/local/stream_protocol.hpp"		//acceptor and socket use
#include "asio/read.hpp"
#include "asio/write.hpp"
#include "asio/post.hpp"
#include "asiobuf.h"

namespace common
{

template<typename HEAD, typename USERDATA>
class LocalStreamServer;

template<typename HEAD, typename USERDATA = bool>
class LocalStreamSession : public std::enable_shared_from_this<LocalStreamSession<HEAD, USERDATA>>
{
	friend class LocalStreamServer<HEAD, USERDATA>;
public:
	USERDATA userdata;
public:
	LocalStreamSession(LocalStreamServer<HEAD, USERDATA> &srv, asio::io_context &ioctx) : server(srv), m_socket(ioctx) {}
	~LocalStreamSession() 
	{
		asio::error_code ec;
		m_socket.shutdown(asio::local::stream_protocol::socket::shutdown_both, ec);
		m_socket.close(ec);
		server.onClose(ip, userdata);
	}

	const std::string & address()
	{
		if (ip.empty() && m_socket.is_open()) {
			asio::error_code ec;
			auto endpoint = m_socket.remote_endpoint(ec);
			if (!ec)
				ip = endpoint.path();//std::move(endpoint.address().to_string());
		}
		return ip;
	}

	//添加发送消息
	void add_sndmsg(std::shared_ptr<KRPC::Buffer> buffer)
	{
		if (!buffer || !(*buffer))
			return;
		sndque.emplace(std::move(buffer));
		if (sndque.size() == 1)
			async_send();
	}
	
	//对于stream而言以下两个函数无效,之所以提供是为了保证所有session提供的接口一致
	bool keep() const {return true;}
	void keep(bool) {}

private:
	LocalStreamServer<HEAD, USERDATA> &server;
	asio::local::stream_protocol::socket m_socket;
	std::string ip;
	HEAD msghead;
	std::queue<std::shared_ptr<KRPC::Buffer>> sndque;

	//设置keepalive心跳机制
	void set_keepalive()
	{
		auto handle = m_socket.native_handle();
	#ifdef __linux__
		int keepAlive = 1; 		// 开启keepalive属性
		int keepIdle = 60;		// 如该连接在60秒内没有任何数据往来,则进行探测 
		int keepInterval = 3; 	//探测时发包的时间间隔为3秒
		int keepCount = 3; 		//探测尝试的次数.如果第1次探测包就收到响应了,则后2次的不再发.
		setsockopt(handle, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepAlive, sizeof(keepAlive));
		setsockopt(handle, SOL_TCP, TCP_KEEPIDLE, (void*)&keepIdle, sizeof(keepIdle));
		setsockopt(handle, SOL_TCP, TCP_KEEPINTVL, (void *)&keepInterval, sizeof(keepInterval));
		setsockopt(handle, SOL_TCP, TCP_KEEPCNT, (void *)&keepCount, sizeof(keepCount));
	#else	//WIN32
		struct tcp_keepalive in_keep_alive = {0};  
	    unsigned long ul_in_len = sizeof(struct tcp_keepalive);  
	    struct tcp_keepalive out_keep_alive = {0};  
	    unsigned long ul_out_len = sizeof(struct tcp_keepalive);  
	    unsigned long ul_bytes_return = 0;  
	  
	    in_keep_alive.onoff = 1;                    /*打开keepalive*/  
	    in_keep_alive.keepaliveinterval = 1000; 	/*发送keepalive心跳时间间隔-单位为毫秒*/  
	    in_keep_alive.keepalivetime = 300000;       /*多长时间没有报文开始发送keepalive心跳包-单位为毫秒*/
	  	WSAIoctl(handle, SIO_KEEPALIVE_VALS, (LPVOID)&in_keep_alive, ul_in_len,   
	                          (LPVOID)&out_keep_alive, ul_out_len, &ul_bytes_return, NULL, NULL);
	#endif
	}

	//异步发送
	void async_send()
	{
		if (sndque.empty())
			return;
		std::shared_ptr<KRPC::Buffer> buffer = sndque.front();
		auto self(this->shared_from_this());
		auto callback = [self, this](asio::error_code ec, std::size_t n){
			if (!ec) {
				server.onSend(self);
				sndque.pop();
				async_send();
			} else {
				server.destroy(self);
			}
		};
		asio::async_write(m_socket, asio::SendBuffer(buffer), callback);
	}
	//异步接收消息
	void async_recv_msg() { async_recv_head(); }

	asio::local::stream_protocol::socket & getSocket() {return m_socket;}

	//异步接收消息头
	void async_recv_head()
	{
		auto self(this->shared_from_this());
		auto callback = [self, this](asio::error_code ec, std::size_t n){
			if (!ec && n == sizeof(msghead)) {
				int bodysize = server.onHead(msghead);
				if (bodysize > 0) {
					asio::RecvBuffer rcvbuf(bodysize + sizeof(msghead), &msghead, sizeof(msghead));
					if (rcvbuf) {
						async_recv_body(rcvbuf);
						return;
					}
				} else if (bodysize == 0) {	//说明本次数据包只有头部没有数据体
					auto buf = std::make_shared<KRPC::Buffer>(sizeof(msghead));
					buf->write((const char *)&msghead, sizeof(msghead));
					if (server.onMessage(std::move(buf), self)) {
						async_recv_head();
						return;
					} 
				}
			}
			server.destroy(self);
		};
		memset(&msghead, 0, sizeof(msghead));
		//asio::async_read(m_socket, asio::buffer(&msghead, sizeof(msghead)), asio::MakeCustomAsyncHandler(callback));
		asio::async_read(m_socket, asio::buffer(&msghead, sizeof(msghead)), callback);
	}

	//异步接收消息体
	void async_recv_body(asio::RecvBuffer rcvbuf)
	{	
		auto self(this->shared_from_this());
		auto callback = [self, this, rcvbuf](asio::error_code ec, std::size_t n){
			if (!ec && server.onMessage(rcvbuf.buffer(), self)) {
				async_recv_head();
			} else {
				server.destroy(self);
			}
		};

		//asio::async_read(m_socket, rcvbuf, asio::MakeCustomAsyncHandler(callback));
		asio::async_read(m_socket, rcvbuf, callback);
	}
};

template<typename HEAD, typename USERDATA = bool>
class LocalStreamServer
{
public:
	using UserData = USERDATA;
	using SessionSPtr = std::shared_ptr<LocalStreamSession<HEAD, USERDATA>>;
	using SessionWPtr = std::weak_ptr<LocalStreamSession<HEAD, USERDATA>>;
	using SessionSet = std::set<SessionSPtr>;
	using iterator = typename SessionSet::iterator;
	using const_iterator = typename SessionSet::const_iterator;

	LocalStreamServer() : m_acceptor(m_ioctx), stopped(false) {}

	virtual ~LocalStreamServer() 
	{
		asio::error_code ec;
		m_acceptor.close(ec);
		stopped = true;
		m_ioctx.stop();
	}
	
	//启动服务器,会阻塞
	void start(const char *filename)
	{
		struct stat st;
		if (stat("/tmp/local", &st) == -1 || S_ISDIR(st.st_mode) == 0) {
			(void) mkdir("/tmp/local", 0777);
		}
		std::string path = "/tmp/local/";
		path += filename;
		if (access(path.c_str(), F_OK) == 0) {
			unlink(path.c_str());
		}
		m_acceptor = asio::local::stream_protocol::acceptor(m_ioctx, asio::local::stream_protocol::endpoint(path));
		accept();
		do {
			try {
				m_ioctx.run();
			} catch (const std::exception &e) {
				//std::cerr << "io_context run occur exception: " << e.what() << std::endl;
			}
			m_ioctx.restart();
		} while (!stopped);
	}
	void start(const std::string &filename) {start(filename.c_str());}

	//销毁客户端session
	void destroy(SessionSPtr pSession) {m_sessions.erase(pSession);}

	//对服务器所连接的所有客户端发布消息
	void publish(std::shared_ptr<KRPC::Buffer> buffer)
	{
		if (!buffer || !(*buffer))
			return;
		auto callback = [buffer, this]{
			for (auto &it : m_sessions) {
				it->add_sndmsg(buffer);
			}
		};
		//asio::post(m_ioctx, asio::MakeCustomAsyncHandler(callback));
		asio::post(m_ioctx, callback);
	}
	//通知其他客户端，这个主要用来对其他客户端发布配置更新消息
	void notify(std::shared_ptr<KRPC::Buffer> buffer, SessionSPtr pSession)
	{
		if (!buffer || !(*buffer))
			return;
		auto callback = [buffer, pSession, this]{
			for (auto &it : m_sessions) {
				if (it != pSession)
					it->add_sndmsg(buffer);
			}
		};
		//asio::post(m_ioctx, asio::MakeCustomAsyncHandler(callback));
		asio::post(m_ioctx, callback);
	}
	//发布一个任务交由LocalStreamServer的ioctx执行
	void post(std::function<void()> task)
	{
		asio::post(m_ioctx, std::move(task));
	}

	//客户端连接之后调用
	virtual void onConnect(SessionSPtr pSession) {}
	//读取到消息头部之后调用,返回后面要接收的消息体size(即去除HEAD),返回-1表示接收的head有误应关闭session,返回0表示本次数据只有head没有body
	virtual int onHead(HEAD &head) = 0;
	//读取到客户端数据之后调用
	virtual bool onMessage(std::shared_ptr<KRPC::Buffer> buffer, SessionSPtr pSession) = 0;
	//发送成功之后调用
	virtual void onSend(SessionSPtr pSession) {}
	//客户端关闭之后调用
	virtual void onClose(const std::string &ip, UserData &data) {}

	iterator begin() {return m_sessions.begin();}
	const_iterator begin() const {return m_sessions.begin();}
	iterator end() {return m_sessions.end();}
	const_iterator end() const {return m_sessions.end();}

private:
	asio::io_context 		m_ioctx;
	asio::local::stream_protocol::acceptor m_acceptor;
	std::set<SessionSPtr> 	m_sessions;
	std::atomic_bool stopped;

	void accept()	//非阻塞接收客户端连接
	{
		SessionSPtr pSession = std::make_shared<LocalStreamSession<HEAD, USERDATA>>(*this, m_ioctx);
		if (!pSession)
			return;
		auto callback = [pSession, this](asio::error_code ec){
			if (!ec) {
				m_sessions.insert(pSession);
				pSession->set_keepalive();
				onConnect(pSession);
				pSession->async_recv_msg();
			}
			accept();
		};
		//m_acceptor.async_accept(pSession->getSocket(), asio::MakeCustomAsyncHandler(callback));
		m_acceptor.async_accept(pSession->getSocket(), callback);
	}
};

}

